"""
案例练习
读取 "orders.txt" 文件，使用 spark 读取文件进行计算
各个城市销售额排名 从大到小
全部城市，有哪些商品类别在售卖
北京市有哪些商品类别在售卖
"""
from pyspark import SparkConf, SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("case")
sc = SparkContext(conf=conf)
# 读取数据
file_rdd = sc.textFile("orders.txt")

# 获取JSON字符串RDD
json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
# 将json字符串转换为字典RDD
json_dict_rdd = json_str_rdd.map(lambda x: json.loads(x))
# print(json_dict_rdd.collect())
# 获取城市销售额
city_money_add = json_dict_rdd.map(lambda x: (x['areaName'], int(x['money'])))
city_money_reduce = city_money_add.reduceByKey(lambda x, y: x + y)

# 计算各个城市销售额排名 从大到小
print(city_money_reduce.sortBy(lambda x: x[1], ascending=False).collect())

# 全部城市，有哪些商品类别在售卖
city_category = json_dict_rdd.map(lambda x: x['category'])
city_category_distinct = city_category.distinct()


city_category_beijing_distinct = json_dict_rdd.filter(lambda x: x['areaName'] == '北京').map(lambda x:x['category']).distinct()
print(city_category_distinct.collect())
print(city_category_beijing_distinct.collect())